package summary

import (
	"EasyCollect/global"
	"database/sql"
	"fmt"
	log "code.google.com/p/log4go"
	"EasyCollect/sqls"
	"errors"
)
/**
	分片项结构体
 */
type ShardingItem struct {
	Value int
}

/**
	获取查询的数据。
 */
func FetchData(task *global.Task,db *sql.DB,sharding *ShardingItem) ([]map[string]interface{}){

	sqlStr,params := querySqlBuildNoSharding(task)

	shardingLog :=""
	if nil!= sharding {
		shardingLog =fmt.Sprintf("sharding: [%v]",sharding.Value)
		sqlStr = addShardingForSql(sqlStr,sharding.Value,task.ShardingColumn)
	}
	log.Info("taskId [%s] taskName [%s] taskCommand [%s] %s querySql [%s] params [%v]",task.TaskId,task.TaskName,task.TaskCommand,shardingLog,sqlStr,params)

	results,err := sqls.GetList(sqlStr,params,db)

	if err != nil {
		errStr := fmt.Sprintf("taskId [%s] taskName [%s] taskCommand [%s] %s sql [%s] params [%v] err [%s]",
			task.TaskId,task.TaskName,task.TaskCommand,shardingLog,sqlStr,params,err)
		log.Error(errStr)
		return make([]map[string]interface{},0)
	}

	//在return数据之前预检验Key的配置是不是正确的
	if len(results)>0 && nil ==results[0][task.SrcKey]{
		errStr := fmt.Sprintf(" sql [%s] key is wrong,your key:[%s],results [%s]",
			sqlStr,task.SrcKey,results)
		log.Error(errStr)
		global.PanicError(errors.New(errStr),task)
	}
	return results
}
/*
	如果配置了targetBeforeSql，则在执行目标库任务时先执行targetBeforeSql
 */
func ExcuseTaskBeforeSql(taskCfg *global.Task, targetDb *sql.DB) error{

	if taskCfg.TargetBeforeSql==""{
		return nil
	}

	affected,err := sqls.ExecSql(taskCfg.TargetBeforeSql,targetDb,make([]interface{},0))
	if err!=nil {
		errStr := fmt.Sprintf("taskId [%s] taskName [%s] taskCommand [%s] TargetBeforeSql [%s] err [%s]",
			taskCfg.TaskId,taskCfg.TaskName,taskCfg.TaskCommand,taskCfg.TargetBeforeSql,err)
		log.Error(errStr)
		return err
	}
	log.Info("taskId [%s] taskName [%s] taskCommand [%s] TargetBeforeSql [%s] rowAffected [%v]",
		taskCfg.TaskId,
		taskCfg.TaskName,
		taskCfg.TaskCommand,
		taskCfg.TargetBeforeSql,affected)
	return nil
}

/*
	如果配置了targetBeforeSql，则在执行目标库任务时先执行targetBeforeSql,此方法为执行带分片的
 */
func ExcuseTaskBeforeSqlSharding(taskCfg *global.Task, targetDb *sql.DB,sharding int) error{

	if taskCfg.TargetBeforeSql==""{
		return nil
	}
	shardingTargetBeforeSql :=addShardingForSql(taskCfg.TargetBeforeSql,sharding,taskCfg.TargetShardingColumn)
	affected,err := sqls.ExecSql(shardingTargetBeforeSql,targetDb,make([]interface{},0))
	if err!=nil {
		errStr := fmt.Sprintf("taskId [%s] taskName [%s] taskCommand [%s] sharding: [%v] TargetBeforeSql [%s] err [%s]",
			taskCfg.TaskId,taskCfg.TaskName,taskCfg.TaskCommand,sharding,shardingTargetBeforeSql,err)
		log.Error(errStr)
		return err
	}
	log.Info("taskId [%s] taskName [%s] taskCommand [%s] sharding: [%v] TargetBeforeSql [%s] rowAffected [%v]",
		taskCfg.TaskId,
		taskCfg.TaskName,
		taskCfg.TaskCommand,
		sharding,
		shardingTargetBeforeSql,affected)
	return nil
}

/**
	执行搬表或者update的任务
 */
func ExcuseInsertOrUpdate(taskCfg *global.Task,srcDb *sql.DB,targetDb *sql.DB,result map[string]interface{}){
	var rowAffected int64
	if taskCfg.TargetType== global.InsertType {
		rowAffected =insertTargetData(taskCfg,targetDb,result)
	}else{
		rowAffected =updateTargetData(taskCfg,targetDb,result)
	}
	//如果配置SrcDelete为true,删除目标源的数据，但同时，任务类型是insert的时候，才删除目标源数据。否则，不会删除src-table中的数据。
	if taskCfg.SrcDelete && taskCfg.TargetType==global.InsertType && rowAffected >0 {
		deleteSrcData(taskCfg,srcDb,result)
	}
}
/**
	删除src-table中的数据
 */
func deleteSrcData(taskCfg *global.Task, srcDb *sql.DB,result map[string]interface{}) {

	sqlStr := buildDeleteSql(taskCfg,result)
	affected,err := sqls.ExecSql(sqlStr,srcDb,make([]interface{},0))
	if err!=nil {
		errStr := fmt.Sprintf("taskId [%s] taskName [%s] taskCommand [%s] deleteSrcSql [%s] err [%s]",
			taskCfg.TaskId,taskCfg.TaskName,taskCfg.TaskCommand,sqlStr,err)
		log.Error(errStr)
		return
	}
	log.Info("taskId [%s] taskName [%s] taskCommand [%s] DeleteSrcSql [%s] rowAffected [%v]",
		taskCfg.TaskId,
			taskCfg.TaskName,
				taskCfg.TaskCommand,
					sqlStr,affected)

}

/**
	向目标库入库数据（搬表或者数据搜集）
 */
func insertTargetData(taskCfg *global.Task, targetDb *sql.DB, result map[string]interface{}) (rowAffected int64) {

	sqlStr,params:= buildInsertSql(taskCfg,result)
	affected,err := sqls.ExecSql(sqlStr,targetDb,params)
	if err!=nil {
		errStr := fmt.Sprintf("taskId [%s] taskName [%s] taskCommand [%s] InsertTargetSql [%s] params [%v] err [%s]",
			taskCfg.TaskId,taskCfg.TaskName,taskCfg.TaskCommand,sqlStr,params,err)
		log.Error(errStr)
		return 0
	}
	log.Info("taskId [%s] taskName [%s] taskCommand [%s] InsertTargetSql [%s] params [%v] rowAffected [%v]",
		taskCfg.TaskId,
		taskCfg.TaskName,
		taskCfg.TaskCommand,
		sqlStr,params,affected)

	return affected
}

/**
	如果配置了targetType="update",则表示更新目标库数据。执行此方法
 */
func updateTargetData(taskCfg *global.Task, targetDb *sql.DB, result map[string]interface{}) (rowAffected int64){
	sqlStr,params:= buildUpdateSql(taskCfg,result)
	affected,err := sqls.ExecSql(sqlStr,targetDb,params)
	if err!=nil {
		errStr := fmt.Sprintf("taskId [%s] taskName [%s] taskCommand [%s] updateTargefSql [%s] params [%v] err [%s]",
			taskCfg.TaskId,taskCfg.TaskName,taskCfg.TaskCommand,sqlStr,params,err)
		log.Error(errStr)
		return 0
	}
	log.Info("taskId [%s] taskName [%s] taskCommand [%s] updateTargetSql [%s] params [%v] rowAffected [%v]",
		taskCfg.TaskId,
		taskCfg.TaskName,
		taskCfg.TaskCommand,
		sqlStr,params,affected)

	return affected
}